package com.kancy.streamplus.channel;

import com.kancy.streamplus.config.DynamicInterfaceBeanResolver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.annotation.Import;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Map;

/**
 * AutoSubscribeChannelPostProcessor
 *
 * @author kancy
 * @date 2020/3/23 0:03
 */
@Import({RetryPollingMessageExceptionHandler.class, EmailNoticePollingMessageExceptionHandler.class})
public class AutoSubscribeChannelPostProcessor extends AbstractPollingMessageListener
        implements MessageHandler, InitializingBean {

    private static final Logger log = LoggerFactory.getLogger(AutoSubscribeChannelPostProcessor.class);

    private final Map<String, AutoSubscribeChannel> channelMap;

    public AutoSubscribeChannelPostProcessor(DynamicInterfaceBeanResolver dynamicInterfaceBeanResolver,
                                             List<PollingMessageExceptionHandler> pollingMessageExceptionHandlers,
                                             Map<String, AutoSubscribeChannel> channelMap) {
        super(dynamicInterfaceBeanResolver, pollingMessageExceptionHandlers);
        this.channelMap = channelMap;
    }

    @Override
    public void afterPropertiesSet() {
        if (!CollectionUtils.isEmpty(channelMap)){
            channelMap.entrySet().stream().forEach(e ->{
                AutoSubscribeChannel subscribeChannel = e.getValue();
                subscribeChannel.input().subscribe(this);
            });
        }
    }

    @Override
    public void handleMessage(Message<?> message) {
        try {
            super.onMessage(message);
        } catch (Exception e) {
            log.info(e.getMessage(), e);
        }
    }

}
